package bbkj.org.cn.window;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import lombok.AllArgsConstructor;
import lombok.Data;

public class WindowHandler {
	private volatile boolean stop;
	private LinkedBlockingQueue<Resource<Object>> queue = new LinkedBlockingQueue<>();
	public void execute() {
		
		//主线程
		while(!stop) {
			
			Resource<Object> resource = queue.poll();
			List<Object> lastDataPoint = resource.getData();
			for(Action<Object> action : resource.getActions()) {
				//负责接受数据
					List<DataPackage<Object>> collectorContainer = new ArrayList<>();
					
					//开线程处理；
					LinkedList<Object> queueTmp = new LinkedList<>();//队列处理
					AtomicInteger currentAdded = new AtomicInteger();
					AtomicInteger times = new AtomicInteger();
					for(Object obj : lastDataPoint) {
						if(queueTmp.size()<action.getSize()) {
							queueTmp.add(obj);
							currentAdded.incrementAndGet();
						}
						if(currentAdded.get()<action.getSlide()) {
							continue;
						}
						times.incrementAndGet();
						try {
							//处理结果放置，堆里面
							Collector<Object> collector = new Collector<Object>() {
								@Override
								public void collect(Object t) {
									collectorContainer.add(new DataPackage<Object>(times.get(),t));
								}
								
							};
							action.apply(queueTmp, collector);
						} catch (Exception e) {
							e.printStackTrace();
						}
						if(queueTmp.size()>=action.getSize()) {
							for(int i=0;i<action.getSlide();i++) {
								queueTmp.removeFirst();
							}
						}
					}
					collectorContainer.sort(new Comparator<DataPackage<Object>>() {
						@Override
						public int compare(DataPackage<Object> o1, DataPackage<Object> o2) {
							return o1.getNum()-o2.getNum();
						}
					});
					lastDataPoint = collectorContainer.stream().map(dp->dp.getData()).collect(Collectors.toList());
			}
			resource.sink().commit(lastDataPoint);
		}
		//结果排序
	}
	interface Collector<T>{
		void collect(T t);
	}
	
	@Data
	@AllArgsConstructor
	class DataPackage<T>{
		private int num;
		private T data;
	}
}
